package common.messages;

import java.util.*;

/**
 * 
 * @author zhou,shi
 * 
 */
public class KVAdminMessage {

	// IP Port Start-index End-index
	// 127.0.0.1 , 50000, 000 ... 000 , 1FF ...FFF
	// 127.0.0.1 , 50001, 200 ... 000 , 3FF ... FFF
	// 127.0.0.1 , 50002, 400 ... 000 , 5FF ... FFF

	private String messageTag = "";
	private String Command = "";
	private List<MetaData> metaList;
	public HashMap<String, String> activeMap;
	private static final char LINE_FEED = 0x0A;
	private static final char RETURN = 0x0D;

	/**
	 *  the ECS has to determine the positions of each server 
	 * within the ring topology to setup the service meta­data.
	 * Once all server ranges
	 *  are determined (i.e., the initial meta­data is complete),  it sends
	 * initialization
	 *  messages that contain the meta­data to all storage servers.
	 */
	public KVAdminMessage() {
		super();
	}

	public KVAdminMessage(HashMap<String, String> activeMap) {
		super();
		this.setMetaList(sortMetaData(activeMap));
	}

	public KVAdminMessage(String command, HashMap<String, String> activeMap) {
		super();
		Command = command;
		this.activeMap = activeMap;
	}

	public String getMessageTag() {
		return messageTag;
	}

	public void setMessageTag(String messageTag) {
		this.messageTag = messageTag;
	}

	public String getCommand() {
		return Command;
	}

	public void setCommand(String command) {
		Command = command;
	}

	public void setMetaList(List<MetaData> metaList) {
		this.metaList = metaList;
	}

	public List<MetaData> getMetaList() {
		return metaList;
	}

	/**
	 * sort the metadata to get the order in hash ring
	 * 
	 * @param activeMap
	 * @return sorted metadata as the order in the hash ring
	 */
	public static List<MetaData> sortMetaData(HashMap<String, String> activeMap) {
		List<MetaData> metaList = new ArrayList<MetaData>();
		List<Map.Entry<String, String>> serverList = new ArrayList<Map.Entry<String, String>>(
				activeMap.entrySet());

		// sort value in hash map
		Collections.sort(serverList,
				new Comparator<Map.Entry<String, String>>() {
					public int compare(Map.Entry<String, String> o1,
							Map.Entry<String, String> o2) {
						return (o1.getValue()).toString().compareTo(
								o2.getValue().toString());
					}
				});
		for (int i = 0; i < serverList.size(); i++) {
			// String id = serverList.get(i).toString();
			String[] ipPort = serverList.get(i).getKey().split(":");
			MetaData metaData = new MetaData();
			if (i == 0) {
				metaData.setHead(serverList.get(serverList.size() - 1)
						.getValue());
			} else {
				metaData.setHead(serverList.get(i - 1).getValue());
			}
			if (i == serverList.size() - 1) {
				metaData.setTail(serverList.get(serverList.size() - 1)
						.getValue());
			} else {
				metaData.setTail(serverList.get(i).getValue());
			}
			metaData.setIp(ipPort[0]);
			metaData.setPort(ipPort[1]);
			metaList.add(metaData);
		}
		return metaList;
	}

	/**
	 * 
	 * marshal metadata to byte array
	 * 
	 * @param km
	 *            input a KVmessage
	 * @return the byte array
	 */
	public byte[] toByteArray(KVAdminMessage kma) {
		messageTag = kma.getMessageTag();
		String PrefixString = messageTag + ";" + kma.getCommand() + ";";
		String meta = PrefixString + metadata2String(kma.getMetaList());
		byte[] bytes = meta.getBytes();
		byte[] ctrBytes = new byte[] { LINE_FEED, RETURN };
		byte[] tmp = new byte[bytes.length + ctrBytes.length];
		System.arraycopy(bytes, 0, tmp, 0, bytes.length);
		System.arraycopy(ctrBytes, 0, tmp, bytes.length, ctrBytes.length);
		return tmp;
	}

	/**
	 * Unmarshal the byte array to KVAdminMessage
	 * 
	 * @param bytes
	 * @return KVMessage
	 */
	public static KVAdminMessage toKVAdminMessage(byte[] bytes) {
		KVAdminMessage kva = new KVAdminMessage();
		String orgiString = new String(bytes).trim();
		String[] metaStrings = orgiString.split(";");
		kva.setMessageTag(metaStrings[0]);
		kva.setCommand(metaStrings[1]);
		String metaString = metaStrings[2];
		// logger.info("The KVserver gets MetaData String which is: "
		// + metaString);

		List<MetaData> metaList = string2MetaData(metaString);
		kva.setMetaList(metaList);
		return kva;
	}

	/**
	 * transform metadata list to string
	 * 
	 * @param metaList
	 * @return string of metalist
	 */
	private static String metadata2String(List<MetaData> metaList) {
		String metaString = "";
		for (MetaData m : metaList) {
			metaString += m.getIp() + ":" + m.getPort() + ":" + m.getHead()
					+ ":" + m.getTail() + ",";
		}
		return metaString;
	}

	/**
	 * tansform metadata string to metadata list
	 * 
	 * @param metaString
	 * @return metadata list
	 */
	private static List<MetaData> string2MetaData(String metaString) {
		List<MetaData> metaList = new ArrayList<MetaData>();
		String[] listSpilit = metaString.split(",");
		for (int i = 0; i < listSpilit.length; i++) {
			String[] metaSpilit = listSpilit[i].split(":");
			MetaData m = new MetaData(metaSpilit[2], metaSpilit[3],
					metaSpilit[0], metaSpilit[1]);
			metaList.add(m);
		}
		return metaList;
	}
}
